/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import org.sss.client.Job;
import org.sss.client.JobEngine;
import org.sss.client.SssClient;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.HashPartitioner2;

public class DataSpreaderApp {
  public static void usage() {
    System.err.println("java DataSpreaderApp commid keysize keycard keycount valsize localonly.");
    System.err.println("  keysize must be >= 4");
    System.err.println("  localonly shoud be `true' or `false'");
    System.exit(-1);
  }

  /**
   * Usage:
   * java DataSpreaderApp commid keysize keycard keycount valsize localonly
   */
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);
    String[] others = client.getArgs();
    if (others.length != 6 || Integer.parseInt(others[1]) < 4) {
      usage();
    }
    String input = others[0];
    int keysize = Integer.parseInt(others[1]);
    int keycard = Integer.parseInt(others[2]);
    int keycount = Integer.parseInt(others[3]);
    int valsize = Integer.parseInt(others[4]);
    boolean localonly = Boolean.parseBoolean(others[5]);

    JobEngine engine = new JobEngine(client);
    try {
      GroupID ci = GroupID.createFromString(input);
      GroupID co = GroupID.createRandom(engine);
      Job.Builder b = engine.getJobBuilder("jobDS", DataSpreader.class)
          .addInput(ci).addOutput(co, Job.partitioner(HashPartitioner2.class));
      Configuration conf = b.getConfiguration();
      conf.setInt("dataspreader.key.size", keysize);
      conf.setInt("dataspreader.key.cardinality", keycard);
      conf.setInt("dataspreader.key.count", keycount);
      conf.setInt("dataspreader.val.size", valsize);
      conf.setBoolean("dataspreader.localonly", localonly);
      @SuppressWarnings("unused")
      Job j = b.build();
      engine.exec();
      System.out.println(co);
    }
    finally {
      engine.dispose();
    }
  }
}
